﻿using System;
using log4net;

namespace Cluster
{
    public abstract partial class ClusterManager
    {
        /// <summary>
        /// Distributed prioritized tasks queue.
        /// Replication tasks have higher priority than balance tasks.
        /// </summary>
        private class ReplicationQueue
        {
            public enum TaskType
            {
                Replicate,
                Balance,
            }

            private class Task
            {
                public Guid OwnerId { get; set; }       // owner guid
                public TaskType Type { get; set; }      // task type
                public string Sequence { get; set; }    // automnatic sequence
            }

            private readonly ILog log = LogManager.GetLogger(typeof(ReplicationQueue));

            public class TaskEventArgs : EventArgs
            {
                public TaskType Type { get; set; }
            }

            private readonly ClusterManager cm;

            private PathWatcher queueWatcher;

            public event EventHandler<TaskEventArgs> TaskDequeued;

            public ReplicationQueue(ClusterManager cm)
            {
                this.cm = cm;
                
                cm.CreateZkPath(ZkQueuePath, true);

                WatchQueue();
            }

            /// <summary>
            /// Adds a new task to the queue
            /// </summary>
            /// <param name="type"></param>
            public void RegisterTask(TaskType type)
            {
                var zpath = string.Format("{0}/{1}_{2}_", ZkQueuePath, cm.uniqueId, type);

                var taskPath = cm.CreateZkNode(zpath, false, true);

                if (log.IsDebugEnabled)
                {
                    log.Debug("Created new task " + taskPath);
                }
            }

            /// <summary>
            /// Setups a watcher on the queue
            /// </summary>
            private void WatchQueue()
            {
                if (queueWatcher == null)
                {
                    queueWatcher = new PathWatcher(OnQueueChange, WatchQueue);
                }

                cm.WatchZkPath(ZkQueuePath, queueWatcher);
            }

            /// <summary>
            /// Called when the queue changes (task is added or removed)
            /// </summary>
            private void OnQueueChange()
            {
                var tasks = cm.GetZkChildren(ZkQueuePath);
                
                // sort task by priority and id
                tasks.Sort((s1, s2) =>
                {
                    var t1 = ParseTask(s1);
                    var t2 = ParseTask(s2);

                    int delta = t1.Type.CompareTo(t2.Type);
                    if (delta == 0)
                    {
                        delta = String.Compare(t1.Sequence, t2.Sequence, StringComparison.Ordinal);
                    }

                    return delta;
                });

                // dequeue if the first task belongs to this node
                if (tasks.Count > 0)
                {
                    var ts = tasks[0];
                    var task = ParseTask(ts);

                    if (task.OwnerId == cm.uniqueId)
                    {
                        if (log.IsDebugEnabled)
                        {
                            log.DebugFormat("Dequeued new {0} task", task.Type);
                        }

                        try
                        {
                            if (TaskDequeued != null)
                            {
                                TaskDequeued(this, new TaskEventArgs
                                {
                                    Type = task.Type
                                });
                            }
                        }
                        finally
                        {
                            if (log.IsDebugEnabled)
                            {
                                log.Debug("Removing task from queue: " + ts);
                            }

                            cm.RemoveZkPath(ZkQueuePath + "/" + ts);
                        }
                    }
                }
            }

            /// <summary>
            /// fills the Task struct by parsing the task name
            /// </summary>
            /// <param name="task">the name</param>
            /// <returns>the task</returns>
            private Task ParseTask(string task)
            {
                var elements = task.Split('_');
                if (elements.Length != 3)
                {
                    throw new ArgumentException("task");
                }

                return new Task
                {
                    OwnerId = Guid.Parse(elements[0]),
                    Type = (TaskType) Enum.Parse(typeof (TaskType), elements[1]),
                    Sequence = elements[2],
                };
            }

            /// <summary>
            /// The Zoo Keeper queue path
            /// </summary>
            private string ZkQueuePath
            {
                get { return "/" + cm.ClusterName + "/replication_queue"; }
            }
        }
    }
}
